package com.wsl.shoppingkill.serviceImpl.mq;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.rabbitmq.client.Channel;
import com.wsl.shoppingkill.common.util.DateUtil;
import com.wsl.shoppingkill.component.request.AbstractCurrentRequestComponent;
import com.wsl.shoppingkill.domain.Order;
import com.wsl.shoppingkill.domain.Sku;
import com.wsl.shoppingkill.mapper.ActivityMapper;
import com.wsl.shoppingkill.mapper.OrderMapper;
import com.wsl.shoppingkill.mapper.SkuMapper;
import com.wsl.shoppingkill.obj.constant.BaseEnum;
import com.wsl.shoppingkill.obj.constant.RabbitMqEnum;
import com.wsl.shoppingkill.obj.constant.RedisEnum;
import com.wsl.shoppingkill.obj.param.AddOrderParam;
import com.wsl.shoppingkill.obj.vo.KillGoodsVO;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.interceptor.TransactionAspectSupport;

import javax.annotation.Resource;
import java.io.IOException;
import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * 订单处理
 *
 * @author : WangShiLei
 * @date : 2021/1/2 12:18 上午
 **/

@Component
@Slf4j
public class OrderDealWith {

    @Resource
    private OrderMapper orderMapper;

    @Resource
    private SkuMapper skuMapper;

    @Resource
    private ActivityMapper activityMapper;


    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    @Resource
    private RabbitTemplate rabbitTemplate;

    @Resource
    private AbstractCurrentRequestComponent abstractCurrentRequestComponent;

    /**
     * 普通订单处理
     * 上文：
     *  public String commonOrder(AddOrderParam addOrderParam) {
     * @param addOrderParam:
     * @param channel:
     * @param message:
     * @author : WangShiLei
     * @date : 2020/11/15 10:55 下午
     **/
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = RabbitMqEnum.Queue.QUEUE_ORDER_COMMON,
                    exclusive = "false", autoDelete = "false", durable = "true",
                    arguments = {
                            @Argument(name = "x-max-length", value = "100000", type = "java.lang.Integer")
                    }),
            exchange = @Exchange(RabbitMqEnum.Exchange.EXCHANGE_ORDER_COMMON),
            key = RabbitMqEnum.Key.KEY_ORDER_COMMON,
            ignoreDeclarationExceptions = "true"
    ))
    @Transactional(rollbackFor = Exception.class)
    public void commonOrder(AddOrderParam addOrderParam, Channel channel, Message message) throws IOException {
        try {
            Order order = new Order();
            Sku sku = skuMapper.selectById(addOrderParam.getSkuId());
            //服务端返回的消息中，头信息中包含了 spring_returned_message_correlation 字段，这个就是服务器发送消息(convertAndSend)
            //时候的 correlation_id(correlationData)，通过消息发送时候的 correlation_id 以及返回消息头中
            // 的 spring_returned_message_correlation 字段值，我们就可以将返回的消息内容和发送的消息绑定到一起，
            // 确认出这个返回的内容就是针对这个发送的消息的。
            long correlationData = Long.parseLong(message.getMessageProperties().getHeaders().get("spring_returned_message_correlation").toString());
            BigDecimal expPrice = sku.getExpPrice();
            if (Objects.isNull(expPrice)) {
                expPrice = new BigDecimal(0);
            }
            order.setStatus(BaseEnum.ORDER_TYPE_NOT_PAY)
                    .setPayPrice(sku.getSellPrice().multiply(new BigDecimal(addOrderParam.getNum())).add(expPrice))
                    .setCreatTime(LocalDateTime.now())
                    .setUpdateTime(LocalDateTime.now())
                    .setNum(addOrderParam.getNum())
                    .setSkuId(addOrderParam.getSkuId())
                    .setUserId(addOrderParam.getUserId())
                    .setOrderTime(LocalDateTime.now())
                    .setId(correlationData);
            orderMapper.insert(order);
            sku.setNum(sku.getNum() - 1).updateById();
            rabbitTemplate.convertAndSend(RabbitMqEnum.Queue.QUEUE_ORDER_DELAY, correlationData, msg -> {
                //过期时间60*5*1000
                msg.getMessageProperties().setExpiration("300000");
                return msg;
            });
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
            if (message.getMessageProperties().getRedelivered()) {
                log.warn("普通订单处理失败，回滚订单，进入死信队列");
                /* 拒绝消息，requeue=false 表示不再重新入队，如果配置了死信队列则进入死信队列 */
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
            } else {
                log.warn("普通订单处理失败，回滚订单，重新入队");
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
            }
            e.printStackTrace();
        }
    }


    /**
     * 秒杀订单处理
     *  上文：
     *      public String killOrder(AddOrderParam addOrderParam) {
     *
     * @param addOrderParam:
     * @param channel:
     * @param message:
     * @author : WangShiLei
     * @date : 2020/11/15 10:55 下午
     * @QueueBinding 注解
     * 1、exchange = @Exchange(value = EventConfig.exchangeGR,
     * 表示接收的交换器名称是EventConfig.exchangeGR，自己根据项目情况灵活配置上。
     * 2、value = @Queue(value = queueName
     * 表示接收的消息队列名称是queueName，自己根据项目情况灵活配置上。
     *  autoDelete:是否是一个临时队列。
     *     true ：当所有的consumer关闭后，自动删除queue。
     *     false：当任意一个consumer启动并创建queue后，如果queue中有消息未消费，无论是否有consumer继续执行，都保存queue。
     * 3、type = ExchangeTypes.FANOUT  【消息发到交换机，交换机根据不同的key 发送到不同的队列】
     *  表示接收消息的类型，分为常用的三种
     *      FANOUT(广播），广播模式，只要连接上，消息就全部通过接收到；  没有任何规则，只要是被该类型得交换器绑定的队列，都可以接收到有提供者发出的消息。
     *      DIRECT(订阅与发布），直连模式，表示消息单独发； 完全匹配规则
     *      TOPIC(主题），订阅模式，根据 routingKey 匹配规则，匹配符合规则的消息数据
     *          服从规则匹配，可以利用通配符，只要符合指定得规则得消费者都可以接收到提供者发出得消息。
     * 4、key=xxx，表示接收消息时候的key，TOPIC模式必须配置上  routingKey, eg: key="*.log.*"  key = "info.#.email.#"
     * 5、消息接收消费后，一定要channel.basicAck(deliveryTag, false)删除消息。
     *
     **/
    @RabbitListener(
        bindings = @QueueBinding(
            value = @Queue(
                    value = RabbitMqEnum.Queue.QUEUE_ORDER_KILL,
                    exclusive = "false", autoDelete = "false", durable = "true",
                    arguments = {@Argument(name = "x-max-length", value = "100000", type = "java.lang.Integer")}
            ),
            exchange = @Exchange(RabbitMqEnum.Exchange.EXCHANGE_ORDER_KILL),
            key = RabbitMqEnum.Key.KEY_ORDER_KILL,
            ignoreDeclarationExceptions = "true"
        )
    )
    public void killOrder(AddOrderParam addOrderParam, Channel channel, Message message) throws IOException {
        try {
            LocalDateTime now = LocalDateTime.now();
            Sku sku = skuMapper.selectById(addOrderParam.getSkuId());
            Order order = new Order();
            long oId = Long.parseLong(message.getMessageProperties().getHeaders().get("spring_returned_message_correlation").toString());
            BigDecimal expPrice = sku.getExpPrice();
            if (Objects.isNull(expPrice)) {
                expPrice = new BigDecimal(0);
            }
            order.setStatus(BaseEnum.ORDER_TYPE_NOT_PAY)
                    .setPayPrice(sku.getSellPrice().multiply(new BigDecimal(addOrderParam.getNum())).add(expPrice))
                    .setCreatTime(now)
                    .setUpdateTime(now)
                    .setNum(addOrderParam.getNum())
                    .setSkuId(addOrderParam.getSkuId())
                    .setUserId(addOrderParam.getUserId())
                    .setOrderTime(now)
                    .setId(oId);

            String key = RedisEnum.GOODS_KILL + addOrderParam.getSkuId();

            orderMapper.insert(order);
            sku.setNum(sku.getNum() - 1).updateById();
            activityMapper.desert(sku.getId(), DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss").format(now));

            rabbitTemplate.convertAndSend(RabbitMqEnum.Queue.QUEUE_ORDER_DELAY, oId, msg -> {
                //过期时间60*5*1000
                msg.getMessageProperties().setExpiration("300000");
                return msg;
            });
            //redisTemplate.opsForValue().set(RedisEnum.USER_KILL_NUM+id+sku.getId(),"1");

            Object o = redisTemplate.opsForValue().get(RedisEnum.GOODS_DOING + sku.getGoodsId());
            if (Objects.nonNull(o)) {
                ObjectMapper objectMapper = new ObjectMapper();
                KillGoodsVO killGoodsVO = objectMapper.convertValue(o, KillGoodsVO.class);
                killGoodsVO.setSum(killGoodsVO.getSum() - 1);
                redisTemplate.opsForValue().set(key + killGoodsVO.getId(), killGoodsVO,
                        DateUtil.distanceSecond(LocalDateTime.now(), killGoodsVO.getEndTime()), TimeUnit.SECONDS);
            }
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            //channel.basicAck(deliveryTag, multiple);
            // consumer处理成功后，通知broker删除队列中的消息，如果设置multiple=true，表示支持批量确认机制以减少网络流量。
            // 例如：有值为5,6,7,8 deliveryTag的投递
            // 如果此时channel.basicAck(8, true);则表示前面未确认的5,6,7投递也一起确认处理完毕。
            // 如果此时channel.basicAck(8, false);则仅表示deliveryTag=8的消息已经成功处理。
        } catch (Exception e) {
            if (message.getMessageProperties().getRedelivered()) {
                log.warn("秒杀订单处理失败");
                /* 拒绝消息，requeue=false 表示不再重新入队，如果配置了死信队列则进入死信队列 */
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
                //channel.basicReject(deliveryTag, requeue);
                // 相比channel.basicNack，除了没有multiple批量确认机制之外，其他语义完全一样。
                // 如果channel.basicReject(8, true);表示deliveryTag=8的消息处理失败且将该消息重新放回队列。
                // 如果channel.basicReject(8, false);表示deliveryTag=8的消息处理失败且将该消息直接丢弃。
                // 参数字段类型：
                // deliveryTag：long - 消息投递的唯一标识，作用域为当前channel
                // multiple：boolean - 是否启用批量确认机制
                // requeue：boolean - 消息处理失败是重新放回队列还是直接丢弃
            } else {
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
                //channel.basicNack(deliveryTag, multiple, requeue);
                // consumer处理失败后，例如：有值为5,6,7,8 deliveryTag的投递。
                // 如果channel.basicNack(8, true, true);  表示deliveryTag=8之前未确认的消息都处理失败且将这些消息重新放回队列中。
                // 如果channel.basicNack(8, true, false); 表示deliveryTag=8之前未确认的消息都处理失败且将这些消息直接丢弃。
                // 如果channel.basicNack(8, false, true); 表示deliveryTag=8的消息处理失败且将该消息重新放回队列。
                // 如果channel.basicNack(8, false, false);表示deliveryTag=8的消息处理失败且将该消息直接丢弃。
                // 链接：https://www.jianshu.com/p/c25444c027c1
            }
            e.printStackTrace();
        }
    }


}
